package com.atguigu.edu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import util.DateFormatUtil;
import util.MyKafkaUtil;

/**
 * 独立访客事务事实表
 */
public class DwdTrafficUniqueVisitorDetail {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点相关设置
       /* env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://39.101.79.248:8020//edu_realtime/checkpoint");
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/
        //TODO 3.从kafka的page_log主题中读取数据,并转化为流
        String topic = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_uv_group";
        DataStreamSource<String> kafkaStrDS
                = env.addSource(MyKafkaUtil.getKafkaConsumer(topic,groupId));
        //TODO 4.对数据类型进行转换 josnStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //TODO 5.对数据按照mid分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(data -> data.getJSONObject("common").getString("mid"));
        //TODO 6.使用状态编程，过滤出独立访客
        SingleOutputStreamOperator<JSONObject> filterDS = keyedDS.filter(
                new RichFilterFunction<JSONObject>() {
                    //声明状态，存放上次访问日期
                    private ValueState<String> lastVisitDateState;

                    //生命周期方法中给状态赋值
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("lastVisitDateState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public boolean filter(JSONObject jsonObj) throws Exception {
                        //独立访客数必然是上页id为空的，所以先把上页id不为空的过滤掉，减少数据量
                        String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        if (StringUtils.isNotEmpty(lastPageId)) {
                            return false;
                        }
                        //接着判断 如果状态中不存在，或者当前的访问日期与状态中的上次访问日期不在同一天，则确定是独立访客。并更新状态
                        String lastVisitDate = lastVisitDateState.value();
                        String curDate = DateFormatUtil.toDate(jsonObj.getLong("ts"));
                        if (StringUtils.isEmpty(lastVisitDate) || !curDate.equals(lastVisitDate)) {
                            lastVisitDateState.update(curDate);
                            return true;
                        }
                        return false;
                    }
                }
        );

        filterDS.print("........");
        //TODO 7.将独立访客写到kafka的主题中
        filterDS.map(JSON::toString).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_unique_visitor_detail"));
        env.execute();
    }
}
